home *** CD-ROM | disk | FTP | other *** search
/ The World of Computer Software / The World of Computer Software.iso / proxy11.zip / DATAFLOW.TXT < prev    next >
Text File  |  1992-12-30  |  5KB  |  93 lines
  1. Proxy v1.2 (tasking) provides a tasking facility using the message
  2. passing paradigm. The basic constructs are tasks and channels.
  3. A task is defined by prefixing the 'task' modifier to a function
  4. definition. Channels are created calling the built-in function of
  5. no arguments: channel(). Other constructs are the start operator
  6. which, when applied to an argument list, activates a task, the
  7. skip operator which terminates a task, (normally, a task only
  8. terminates when it reaches the end of the task body), an input
  9. operator ? which is applied to a channel and an input variable, and
  10. an output operator ! which is applied to a channel and an
  11. expression. The syntax of the input/output operators is:
  12.  
  13. channel-id ? identifier     The current task is blocked until
  14.                             a message is available in channel-id.
  15.                             Then the message is stored in
  16.                             identifier.
  17.  
  18. channel-id ! expression     Expression is evaluated and the
  19.                             value is transmitted in channel-id.
  20.  
  21. The use of these operators is shown in a simple dataflow
  22. diagram where tasks correspond to process nodes, and channels
  23. correspond to data flows.
  24.  
  25.  
  26.          ___________      _________      __________
  27.          |         | trans|       |group |        |
  28.          | Source  |______| Group |______|  Sink  |
  29.          |         |      |       |      |        |
  30.          |_________|      |_______|      |________|
  31.  
  32.  
  33. There are three process nodes, Source, Group and Sink. Source produces
  34. a stream of records or transactions, where each record consists of
  35. a partno (which may be considered to be an integer for simplicity)
  36. and a quantity. Quantities may be positive or negative, and may be
  37. thought of as representing receipts or issues into or out of a
  38. warehouse. Source obtains its data from a file, where each line
  39. of the file consists of the partno and quantity. Further, the lines
  40. of the file are considered to be sorted on the partno. The purpose
  41. of Group is to collect all the records of the same partno into an
  42. aggregate data structure and pass these aggregates to Sink. The
  43. purpose of Sink is to produce a report giving for each partno the
  44. net issue or receipt of that partno. The process nodes are represented
  45. by tasks and the data flows by channels. We show below the Proxy program
  46. (with appropriate comments), to execute the dataflow diagram.
  47.  
  48. ch1=channel();         // channels ch1 and ch2 are declared
  49. ch2=channel();
  50. eos="eos";             // the end of stream indicator is defined
  51.  
  52. // reduce adds algebraicly all the integers in a sequence.
  53.  
  54. reduce(x) {if(x==[]) return 0;else return hd x+reduce(tl x);};
  55.  
  56. struct trans {partno,qty;};  // trans is defined with components partno
  57.                              // and quantity
  58. task source(out) {p=open("srce.in","r");readln(p,partno,qty);
  59.     while(!eof) {
  60.                  out!new trans(partno,qty); // creates a transaction record
  61.                  readln(p,partno,qty);      // from a file and sends it
  62.                 }                           // on the output channel
  63.                 out!eos;                    // at end of file, sends
  64.                  };                         // the end of stream indicator
  65.  
  66. task group(inp,out;x,y,g,p) {g=[];inp?x;p=x.partno;y=p;
  67.     while (x!=eos) {if(y==p)              // concatenates a record to the
  68.                      {g=g conc [x];inp?x; // sequence as long as partno
  69.                       if(x==eos)          // is the same, if input is eos
  70.                        {out!g;out!eos;skip;} // outputs the sequence, eos
  71.                         p=x.partno;}         // and terminates the task
  72.                     else {out!g;g=[];y=p;}   // if partno is different,
  73.                     }                        // outputs the sequence,
  74.                             };               // initializes new sequence
  75.                                              // and partno
  76.  
  77. task sink(inp;x,y) {inp?x;                    // inputs a sequence
  78.                     while(x!=eos)
  79.                      {y=[z.qty:z<-x];   // y is a sequence of qty's
  80.                       z=hd x;           // from input, z is the first record
  81.                       writeln("partno",z.partno,"net qty",reduce(y));
  82.                     inp?x;}
  83.                    };
  84.  
  85. task main() {start source(ch1);start group(ch1,ch2);start sink(ch2);};
  86.  
  87. The main task starts the tasks source, group and sink by applying them
  88. to channel variables, where a particular channel variable serves both as
  89. the output of one task  and the input to another. Execution is initiated
  90. by invoking the main task:
  91.  
  92. main();
  93.